package com.lh.marketing;

import lombok.extern.slf4j.Slf4j;
import org.apache.commons.lang3.StringUtils;
import org.springframework.data.redis.core.StringRedisTemplate;
import org.springframework.stereotype.Component;
import org.springframework.util.CollectionUtils;

import javax.annotation.Resource;
import java.util.Set;
import java.util.concurrent.LinkedBlockingQueue;
import java.util.concurrent.ThreadPoolExecutor;
import java.util.concurrent.TimeUnit;

/**
 * 自定义的任务操作
 */
@Slf4j
@Component
public class DelayQueueTask {

    @Resource
    StringRedisTemplate stringRedisTemplate;

    //锁
    private final Object lock = new Object();

    private final Object waitLock = new Object();

    protected  String getDelayKey(){
        return "task_delay_key_";
    }

    protected  String getQueueKey(){
        return "task_queue_key_";
    }

    protected  String getLockKey(){
        return "task_lock_key_";
    }

    protected  StringRedisTemplate getStringRedisTemplate(){
        return stringRedisTemplate;
    }

    public void createTask(Long executeTime, String value){
        //任务进来直接丢到zSet中
        getStringRedisTemplate().opsForZSet().add(getDelayKey(),value+":"+executeTime,executeTime);
        //唤醒线程
        synchronized (lock){
            lock.notifyAll();
        }
    }

    public void consume(){
        ThreadPoolExecutor pool = getThreadPool();
        int maxPoolSize = pool.getMaximumPoolSize();
        new Thread(() -> {
            while (true){
                try {
                    //如果当前活跃线程达到了最大线程数的80%，则暂停新增线程
                    if(maxPoolSize * 0.8 < pool.getActiveCount()){
                        //先暂时休息一下
                        synchronized (waitLock){
                            try {
                                waitLock.wait(1000L);
                                continue;
                            } catch (InterruptedException ignored) {
                                continue;
                            }
                        }
                    }
                    String value = consumeOne();
                    if(StringUtils.isBlank(value)){
                        continue;
                    }
                    String[] split = value.split(":");
                    pool.execute(() -> execute(split[0],Long.valueOf(split[1])));
                }catch (Exception e){
                    log.error("任务队列消费异常",e);
                }
            }
        }).start();
    }

    /**
     * 具体任务执行
     * @param value 队列中的任务的值
     */
    protected  void execute(String value,Long time){
        log.info("执行器队列任务执行:{}",value);
    };

    /**
     * 线程池配置
     */
    protected ThreadPoolExecutor getThreadPool(){
        return new ThreadPoolExecutor(5,20,30,TimeUnit.SECONDS,new LinkedBlockingQueue<>(1000));
    }

    protected String consumeOne(){
        return getStringRedisTemplate().boundListOps(getQueueKey()).leftPop(30, TimeUnit.SECONDS);
    }

    protected void start(){
        StringRedisTemplate stringRedisTemplate = getStringRedisTemplate();
        new Thread(() -> {
            while (true){
                try {
                    long now = System.currentTimeMillis();
                    Set<String> values = stringRedisTemplate.opsForZSet().range(getDelayKey(), 0, 1);
                    if(CollectionUtils.isEmpty(values)){
                        synchronized (lock){
                            try {
                                lock.wait(3000L);
                                continue;
                            } catch (InterruptedException e) {
                                e.printStackTrace();
                                continue;
                            }
                        }
                    }
                    String value = values.iterator().next();
                    String[] split = value.split(":");
                    String id = split[0];
                    long time = Long.parseLong(split[1]);
                    if(now < time){
                        synchronized (lock){
                            try {
                                lock.wait(time - now);
                                continue;
                            } catch (InterruptedException e) {
                                e.printStackTrace();
                            }
                        }
                    }
                    //将任务放入待执行队列中
                    Boolean aBoolean = stringRedisTemplate.opsForValue().setIfAbsent(getLockKey(), "0", 3, TimeUnit.SECONDS);
                    if(aBoolean==null || !aBoolean){
                        continue;
                    }
                    Long removeCount = stringRedisTemplate.opsForZSet().remove(getDelayKey(), value);
                    if(removeCount==null || removeCount==0){
                        continue;
                    }
                    stringRedisTemplate.opsForList().rightPush(getQueueKey(),value);
                    stringRedisTemplate.delete(getLockKey());
                }catch (Exception e) {
                    log.error("任务队列扫描异常",e);
                }
            }
        }).start();
    }
}
